package com.atguigu.flink.exec1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.streamapi.source.Demo4_CustomSource;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2023/1/29
 * 
 *      ①读数据
 *      ②把原始数据封装为计算的类型
 *      ③计算
 *      ④输出
 *              print
 */
public class Demo3_ChannelStats
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读数据  {"channel":"MI","behavior":"uninstall","userId":46,"timestamp":1674973100368}
        env.addSource(new Demo4_CustomSource.MySource())
           .map(new MapFunction<String, Tuple2<String,Integer>>()
           {
               @Override
               public Tuple2<String,Integer> map(String line) throws Exception {

                   //本质就是一个Map
                   JSONObject jsonObject = JSON.parseObject(line);

                   return Tuple2.of(
                       jsonObject.getString("channel") + "_" + jsonObject.getString("behavior"),
                       1
                   );

               }
           })
           .keyBy(t -> t.f0)
           .sum(1)
           .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        
    }
}
